package com.ly.kafka.utils;


import com.alibaba.fastjson.JSON;
import com.ly.kafka.empty.Order;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

//生产者API
public class ProducerDemo {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Properties properties = new Properties();
        //集群方式连接以都好隔开
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.23.102:9092,192.168.23.103:9092,192.168.23.105:9092");
        //发出消息持久化机制参数
        //(1) acks=0：表示发送者(producer)不需要等待broker确认收到消息的回复，就可以继续发送下一条消息，性能最高，容易丢失消息
        //(2) acks=1：只好要等待leader已经成功将数据写入本地log,但不需要等待所有的follower是否写入成功。就可以发送下一条信息
                      //如果leader挂掉了 还没来得级同步给follower中 造成数据丢失
        //(3) acks=-1或all ：这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)，
                            //假如有三个分区 配置文件属性min.insync.replicas配置了两个，两个分区日志写入完毕后就会告诉客户端消息发送从成功
                            // 这种策略安全级最高只要保证一个备份存活
                            //存活，数据就不会丢失，一般跟设计到跟钱打交道的业务使用这种配置
        //一般为1就行
        properties.put(ProducerConfig.ACKS_CONFIG,"1");

        //发送失败会重试，默认间隔是100ms 重试既能保证消息发送的可靠性，但也可能造成消息发送重复，比如网络抖动，所以需要
          //在接收者那边做好消息接收的
        properties.put(ProducerConfig.RETRIES_CONFIG,3);

        //重试间隔的设置  300ms后在发送  可能会影响消费的乱序 是否启用该参数需要视业务而定
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,300);

        //设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发给本地缓冲区，可以提高消息发送的性能，默认值是33554432 既32MB
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);

        //kafka本地线程会从缓冲区内取数据，批量发送到broker
        //设置批量发送消息的大小，默认值是16384 既16KB ,意思就是一个batch满了16KB就会发送出去 如果不满就会等待够16kb再发
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);

        //默认值是0 意思是消息必须立刻发送出去，但是会影响性能
        //一般设置在100毫秒左右，就是说这个消息发送完后会进入本地的一个batch，如果100毫秒内，这个batch满了16KB就会随着batch一起发送出去
        //如果100毫秒内，batch没满那么也必须把消息发送出去，不能让消息的发送延迟时时间太长
        properties.put(ProducerConfig.LINGER_MS_CONFIG,100);

        //把发送的key从字符串序列化为字符数组
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息的value从字符串序列化为字符数组
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //设置自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.ly.kafka.utils.CustomizePartitioner");


        Producer<String,String> producer = new KafkaProducer<String, String>(properties);

        int msgNum = 5;
        CountDownLatch countDownLatch = new CountDownLatch(msgNum);

        for (int i=0;i<=msgNum;i++){
            Order order = new Order(i,100+i,1,1000.00);
            //指定发送分区 第0个分区里面发
            ProducerRecord<String,String> producerRecord = new ProducerRecord<String,String>("hello-kraft1",
                    0,order.getOrderId().toString(), JSON.toJSONString(order));
            //等待消息发送成功的同步阻塞方法 用的不多
//            RecordMetadata producerMetadata = producer.send(producerRecord).get();
//            System.out.println("异步消息发送结果：topic-"+producerMetadata.topic()+"|partition-"+
//                    producerMetadata.partition()+"|offset-"+producerMetadata.offset());

            //异步发送消息 不用等到消息发送到服务端后 在继续往下执行业务逻辑 用的比较多
            producer.send(producerRecord, new Callback() {
                @Override
                //RecordMetadata 封装发送消息成功后的一些信息 例如分区等
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e!=null){
                        System.out.println("发送消息失败："+e.getStackTrace());
                    }
                    if(recordMetadata!=null){
                        System.out.println("异步消息发送结果：topic-"+recordMetadata.topic()+"|partition-"+
                                recordMetadata.partition()+"|offset-"+recordMetadata.offset());
                    }
                    countDownLatch.countDown();
                }
            });
            //继续往下走的业务...

        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.close();
    }




}
